package com.yanduo.report

import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 构造投放的地域分布统计
  * 实现方式：
  *   Spark sql实现
  *  知识点:序列化优化  可以参考:https://www.jianshu.com/p/8ccd701490cf
  *课程地址:https://www.bilibili.com/video/BV1F4411i7jK?p=18
  * @author Gerry chan
  * 2020/5/3 9:18
  * @version 1.0
  */
object AreaAnalyseRpt {
  def main(args: Array[String]): Unit = {
    //0 校验参数个数
    if (args.length !=2) {
      println(
        """
          |cn.dmp.report.AreaAnalyseRpt
          |参数：
          | logInputPath
          |
        """.stripMargin
      )
    }

    //1 接受程序参数
    val Array(logInputPath) = args

    //2 创建sparkconf-》sparkContext
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    //RDD序列化到磁盘，worker 与worker之间的数据传输
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)

    //在Spark2.0中只要创建一个SparkSession就够了，
    //SparkConf、SparkContext和SQLContext都已经被封装在SparkSession当中.
    //SparkSession统一了sqlcontext和hivecontext
    val spark = new SparkSession(sc)


    //dataframe -> table
    //读取parquet文件
    val parquetData:DataFrame = spark.read.parquet(logInputPath)
    parquetData.createOrReplaceTempView("log")

    //spark1.x写法
    //parquetData.registerTemplate("log")

    //业务逻辑
    val result:DataFrame = spark.sql(
      """
        |select
        |  provincename, cityname,
        |  sum(oxox(requestmode=1 and processnode>=1)) rawReq,  -- 原始请求
        |  sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) effReq, -- 有效请求
        |  sum(case when requestmode=1 and processnode=3 then 1 else 0 end) adReq, -- 广告请求
        |  sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0 then 1 else 0 end) rtbReq, -- 参与竞价数
        |  sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) succReq, -- 竞价成功数
        |  sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) adShow, -- 展示数
        |  sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) adClick, --点击数
        |  sum(case when iseffective=1 and isbilling=1 and iswin=1 then winprice/1000 else 0 end) xf, -- 广告成本
        |  sum(case when iseffective=1 and isbilling=1 and iswin=1 then adpayment/1000 else 0 end) cb -- 广告消费
        |from logs
        |group by provincename,cityname
      """.stripMargin)


    // 知识点：使用com.typesafe 加载配置文件，
    // 查找顺序：application.conf --》 application.json -->application.properties
    val config: Config = ConfigFactory.load()
    val props: Properties = new Properties()
    props.setProperty("user", config.getString("jdbc.user"))
    props.setProperty("password", config.getString("jdbc.password"))

    //将结果写入mysql ,config.getString("jdbc.tableName").split(",")(0) 取第一个表
    result.write.mode(SaveMode.Append)
      .jdbc(config.getString("jdbc.url"), config.getString("jdbc.tableName").split(",")(0), props)

    spark.stop()
  }

}
